/**一、消费流程：
 * 1.构造Consumer的实例
 * 2.设置必要属性：如NamesrvAddr、MessageModel、ConsumeFromWher
 * 3.订阅Topic和Tag/SQL Filter
 * 4.设置监听器（2种），在其中处理收到的消息---PushConsumer
 * 5.启动consumer
 * 6.拉取消息---PullConsumer
 * 7.手动提交---PullConsumer
 *
 *Consumer有2种，其下有不同消费方式
 * 一、PushConsumer（DefaultMQPushConsumer）：类似异步接收
 * 1.并行消费：MessageListenerConcurrently
 * 2.顺序消费：MessageListenerOrderly
 *
 * 二、PullConsumer（DefaultLitePullConsumer）：类似同步接收
 * 1.订阅式：订阅后主动拉取消息
 * 2.分配式：直接指定队列然后拉取其中消息，手动提交
 *
 * 消费后返回状态有两种：
 * 1.ConsumeConcurrentlyStatus.CONSUME_SUCCESS/ConsumeOrderlyStatus.SUCCESS
 * 2.ConsumeConcurrentlyStatus.RECONSUME_LATER/ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT
 */

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.utils.MessageUtil;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public class DemoConsumer {

    public static void pushConcurrentConsumer() throws InterruptedException, MQClientException {
        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_consumer_group1");

        // Specify name server addresses.
        consumer.setNamesrvAddr("localhost:9876");

        // 订阅Topic和Tag Filter
        consumer.subscribe("TopicA", "*");

         /*SQL Filter，仅提供了一些基本的语法，broker中要配置enablePropertyFilter=true且仅push模式的消费者才能用
             数值比较： >, >=, <, <=, BETWEEN, =;
             字符比较： =, <>, IN;
             IS NULL or IS NOT NULL;
             逻辑： AND, OR, NOT;
             TRUE or FALSE
         */
//        String sql = "TAGS in ('TagA', 'TagB') and color = 'blue' and num > 5";
//        consumer.subscribe("TopicA", MessageSelector.bySql(sql));

        //设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费，如果非第一次启动，那么按照上次消费的位置继续消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        //集群模式或广播模式（MessageModel.BROADCASTING）
        consumer.setMessageModel(MessageModel.CLUSTERING);

        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                try {
                    for (MessageExt messageExt : msgs) {

                        String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        //输出消息内容
                        System.out.println("[  ]Received: msgId --- " + messageExt.getMsgId() + ",  msgBody ---" + messageBody);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    //稍后再试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                //消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //Launch the consumer instance.
        consumer.start();

        System.out.printf("Consumer Started.%n");
    }

    //同一个MessageQueue里面的消息要顺序消费，不能并发消费;同一个Topic的不同MessageQueue是可以同时消费的
    public static void pushOrderedConsumer() throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_consumer_group2");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setNamesrvAddr("localhost:9876");
        // 设置定于的主题和 tag(必须显示指定 tag)
        consumer.subscribe("OrderedTopic", "tagA || tagB || tagC || tagD || tagE");
        consumer.setMessageListener(new MessageListenerOrderly() {
            AtomicLong consumeTimes = new AtomicLong(0);

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                //设置是否自动提交: 默认自动提交, 提交之后消息就不能够被再次消费.非自动提交时, 消息可能会被重复消费
                context.setAutoCommit(true);
                try {
                    for (MessageExt msg : msgs) {
                        System.out.println("[Ordered]Received:num=" + this.consumeTimes.get() + ", queueId=" + msg.getQueueId() + ", Keys=" + msg.getKeys() + ", value=" + new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET));
                    }
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }

                this.consumeTimes.incrementAndGet();
                if ((this.consumeTimes.get() % 2) == 0) {
                    return ConsumeOrderlyStatus.SUCCESS;
                } else if ((this.consumeTimes.get() % 3) == 0) {
                    return ConsumeOrderlyStatus.ROLLBACK;
                } else if ((this.consumeTimes.get() % 4) == 0) {
                    return ConsumeOrderlyStatus.COMMIT;
                } else if ((this.consumeTimes.get() % 5) == 0) {
                    context.setSuspendCurrentQueueTimeMillis(3000);
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                // 非主动提交的时候, SUCCESS 不会导致队列消息提交, 消息未提交就可以被循环消费
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
    }

    //订阅方式,主动拉取消息，其它与pushConsumer类似，支持消息队列负载均衡
    public static void pullConsumer1() throws InterruptedException, MQClientException {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("my_consumer_group2");
        //指定 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        //设置 Consumer 第一次启动时从队列头部开始消费还是队列尾部开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //订阅指定 Topic 下的所有消息
        consumer.subscribe("TopicA", "*");
        // 消费者对象在使用之前必须要调用 start 初始化
        consumer.start();
        try {
            //循环开始消费消息
            while (true) {
                //从consumer中获取消息
                List<MessageExt> list = consumer.poll();
                //消息处理
                if (list != null) {
                    for (MessageExt ext : list) {
                        try {
                            System.out.println(new String(ext.getBody(), RemotingHelper.DEFAULT_CHARSET));
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        } finally {
            // 一旦生产者实例不再被使用则将其关闭，包括清理资源，关闭网络连接等
            consumer.shutdown();
        }
    }

    //分配方式,支持自定义需消费的消息队列，不支持负载均衡
    public static void pullConsumer2() throws InterruptedException, MQClientException {
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("LitePullConsumer");
        // 设置服务器地址
        litePullConsumer.setNamesrvAddr("localhost:9876");
        // 关闭自动提交偏移量
        litePullConsumer.setAutoCommit(false);
        // 启动消费者
        litePullConsumer.start();
        // 获取 Topic 的所有 Queue
        Collection<MessageQueue> mqSet = litePullConsumer.fetchMessageQueues("TopicA");
        List<MessageQueue> list = new ArrayList<>(mqSet);
        //将要订阅的 queue 加入到 assignList
        List<MessageQueue> assignList = new ArrayList<>();
        for (int i = 0; i < list.size() / 2; i++) {
            assignList.add(list.get(i));
            System.out.println(list.get(i));
        }
        //手动为消费者分配一个消息队列列表
        litePullConsumer.assign(assignList);

        //seek方法方便用户重置消费位点，10是每次拉取的最大消息数的缺省值
        litePullConsumer.seek(assignList.get(0), 10);
        try {
            while (true) {
                //消息拉取
                List<MessageExt> messageExts = litePullConsumer.poll();
                if (messageExts != null) {
                    for (MessageExt ext : messageExts) {
                        try {
                            System.out.println(new String(ext.getBody(), RemotingHelper.DEFAULT_CHARSET));
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    }
                }
                //手动提交，否则消息始终在队列中，会被重复消费
                litePullConsumer.commitSync();
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }

    //rocketmq-4.7.0推出的“Request-Reply”特性来支持RPC调用，
    public static void rpcConsumer() throws InterruptedException, MQClientException {
        String producerGroup = "rpc_producer_group";
        String consumerGroup = "rpc_consumer_grou";
        String topic = "RequestTopic";

        // create a producer to send reply message
        DefaultMQProducer replyProducer = new DefaultMQProducer(producerGroup);
        replyProducer.setNamesrvAddr("localhost:9876");
        replyProducer.start();

        // create consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setNamesrvAddr("localhost:9876");

        // recommend client configs
        consumer.setPullTimeDelayMillsWhenException(0L);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                for (MessageExt msg : msgs) {
                    try {
                        System.out.printf("[RPC]received: %s%n", new String(msg.getBody()));
                        byte[] replyContent = "reply message ------.".getBytes();
                        // 根据收到的msg，用MessageUtil来制作回复消息，其中包含了发送着的id
                        Message replyMessage = MessageUtil.createReplyMessage(msg, replyContent);

                        // 用普通的producer发送回复消息，会先发到统一的Reply_Topic中，然后再推送给producer
                        SendResult replyResult = replyProducer.send(replyMessage, 3000);
                        System.out.printf("[RPC]replyResult:  %s %n", replyResult.toString());
                    } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.subscribe(topic, "*");
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}


